package com.lrd.rmq514version.rmq.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;


@Component
public class MyDelayTimeMsgProducer {

    private static final Logger logger = LoggerFactory.getLogger(MyDelayTimeMsgProducer.class);
    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    @Value("${apache.rocketmq.namesrvAddr}")
    private String nameServer;

    @Value("${apache.rocketmq.delay-time-producer.topic}")
    private String delayTimeTopic;

    @Value("${apache.rocketmq.delay-time-producer.group}")
    private String group;

    @Value("${apache.rocketmq.delay-time-producer.tag}")
    private String tag;

    private final DefaultMQProducer producer = new DefaultMQProducer();

    /**
     * 初始化
     */
    @PostConstruct
    public void start() {
        try {
            logger.info("MyDelayTimeMsgProducer|starting...");
            producer.setNamesrvAddr(nameServer);
            producer.setProducerGroup(group);
            //instanceName用来生产clientId【10.0.16.125@test_delay_time_message_producer_instance_01】
            producer.setInstanceName("test_delay_time_message_producer_instance_01");
            producer.start();
            logger.info("MyDelayTimeMsgProducer|生产者启动成功");
        } catch (MQClientException e) {
            logger.error("MyDelayTimeMsgProducer|生产者启动异常!|namesrvAddr={}", nameServer, e);
        }
    }

    /**
     * 发送延时消息
     *
     * @param message 消息体
     * @param
     */
    public boolean sendDelayTimeMessage(String message, String key, Long delayTimeMs) {
        try {
            Message mqMsg = new Message(delayTimeTopic, tag, key, message.getBytes(RemotingHelper.DEFAULT_CHARSET));
            mqMsg.setDelayTimeMs(delayTimeMs);
            SendResult sendResult = producer.send(mqMsg);
            String sendTime = LocalDateTime.now().format(formatter);
            if (sendResult != null && sendResult.getSendStatus() == SendStatus.SEND_OK) {
                logger.info("延时消息[发送]成功,发送时间={},延时时间={}(ms),msgKey={},msgId={},message={}", sendTime, delayTimeMs, key, sendResult.getMsgId(), message);
                return true;
            }
            logger.error("发送延时消息error,sendResult={}", sendResult);
            return false;
        } catch (Exception e) {
            logger.error("MyDelayTimeMsgProducer|sendMessage发送异常|", e);
            return false;
        }
    }


    @PreDestroy
    public void stop() {
        producer.shutdown();
        logger.info("MQ：关闭MyDelayTimeMsgProducer生产者");
    }
}
